package com.lin.reactor;

import org.junit.jupiter.api.Test;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

/**
 * @author: linjiang
 * @date: 2025/5/4
 */
public class ReactorFlux {

    @Test
    public void testFluxJust() {
        Flux<Integer> complete = Flux.just(1, 2, 4, 3, 4, 8, 5, 0)
                .doOnRequest(item -> System.out.println("request:" + item))
                .doOnNext(item -> System.out.println("item:" + item))
                .map(item -> item * 2)
                .doOnNext(item -> System.out.println("item1:" + item))
                .doOnComplete(() -> System.out.println("complete"));

        // 订阅
        complete.subscribe();
    }

    @Test
    public void testFluxFromIterable() {
        Flux<Integer> flux = Flux.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
                .log();

        // 订阅
//        flux.subscribe();
        // 订阅时，对每个产生的值 做处理
//        flux.subscribe(i -> System.out.println("i:" + (i * 10)));
        // 订阅时，对每个产生的值 做处理，并出现异常时，触发回调机制
//        flux.subscribe(i -> System.out.println("i:" + (10 / i)), err -> {
//            System.out.println("err:" + err);
//        });

        flux.subscribe(i -> System.out.println("i:" + (10 >> i)), err -> {
            System.out.println("err:" + err);
        }, () -> System.out.println("complete"));
    }

    @Test
    public void testDisposable() throws InterruptedException {
        // 1.这个发布者会每个1s发送一个数据
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1))
                .log()
                // 6.一旦取消信号发出，这里的取消回调会被执行。
                .doOnCancel(() -> System.out.println("订阅被取消。。。"));

        // 2.每个订阅者都会实现一个取消接口并返回
        Disposable disposable = flux.subscribe();


        // 3.通过它能触发取消信号。让发布者不在产生元素
        TimeUnit.SECONDS.sleep(5);
//        disposable.dispose();
        // 扩展： 将多个Disposable组合起来，一旦触发关闭，所有的发布者都会被取消。
        Disposables.composite(disposable).dispose();
    }

    @Test
    public void testBaseSubscribe() {
        CustomBaseSubscribe<Integer> baseSubscribe = new CustomBaseSubscribe<>();
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5)
                .log();

        flux.subscribe(baseSubscribe);

    }

    @Test
    public void testBackpressureAndRequestReshape() throws IOException {
        /*
         * 解决背压和请求重塑的问题
         *   1.buffer缓冲区。当订阅者无法快速处理发布者发送的数据时，就会将数据缓存起来。降低订阅者的压力
         *   2.请求重塑。当订阅者无法处理数据时，就限制发送给发布者的请求。
         * */

        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1))
                .log()
                .limitRate(10);

        flux.subscribe();

        System.in.read();
    }

    @Test
    public void testFluxGenerate() {
        // 按需生成同步序列并且每次只能生成一个元素。多个会报错。
        Flux<Object> flux = Flux.generate(
                () -> 0, // 关联初始状态
                (state, sink) -> { // 状态，手动发送器
                    sink.next("9 * " + state + " = " + state * 9);
                    if (state == 10) {
                        sink.complete(); // 手动完成
                    }

                    return state + 1;
                }
        );

        // 还是只有订阅了，才会执行。
        flux.subscribe(System.out::println);
    }


    @Test
    public void testAyncMultiThreadCreate() {
        /*
         * 测试异步多线程创建数据
         *   多个线程创建数据，并将数据合并为一个flux流。
         * */
        Flux<Object> flux = Flux.create(sink -> {
            for (int j = 0; j < 5; j++) {
                new Thread(() -> {
                    for (int i = 0; i < 10; i++) {
                        sink.next(Thread.currentThread().getName() + "发送" + i);
                    }
                }).start();
            }
        }).log();

        flux.subscribe();

    }


    @Test
    public void testAyncSingleThreadPush() {
        /*
         * 异步单线程模式
         * */
        // 这段代码按理说是会抛出异常的。
        Flux<Object> flux = Flux.push(fluxSink -> {
            Runnable runnable = () -> fluxSink.next("发送" + Thread.currentThread().getName());
            Thread thread1 = new Thread(runnable);
            Thread thread2 = new Thread(runnable);
            thread1.start();
            thread2.start();

            try {
                thread1.join();
                thread2.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).log();

        flux.subscribe(System.out::println);

    }

    @Test
    public void testSchedulers() throws IOException {
        new Thread(() -> {
            // 源头创建是一个线程
            Flux<Integer> flux = Flux.range(1, 10);

            // 在t1线程中工作。
            flux.log().subscribe();

            // 在t2线程中工作。
            new Thread(() -> {
                // 订阅和操作是另外一个线程。
                flux.map(item -> item * 2)
                        .log()
                        .subscribe();
            }, "t2").start();
        }, "t1").start();
        // 结论： 大多数的操作是继上一个操作的的thread中执行的。

        System.in.read();
    }

    @Test
    public void testSchedulers01() throws IOException {
        Flux<Integer> flux = Flux.range(1, 10);

        // 不做调度，直接在调用者线程中执行。
        flux.map(item -> item * 2)
                .subscribeOn(Schedulers.immediate())
                .log()
                .subscribe();


        // 单线程调度器，所有任务都在一个线程中顺序执行。
        flux.map(item -> item % 2)
                .subscribeOn(Schedulers.newSingle("t1"))
                .log()
                .subscribe();

        // 弹性线程调度器。将任务分配给弹性线程池。使其阻塞线程不要阻塞主线程。
        flux.map(item -> {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return item * 2;
                })
                .subscribeOn(Schedulers.boundedElastic())
                .log().subscribe();
        System.out.println(1111111);

        // 并行调度器。
        flux.map(item -> item * 2)
                .subscribeOn(Schedulers.newParallel("t2"))
                .log()
                .subscribe();

        Flux.interval(Duration.ofSeconds(1), Schedulers.newSingle("t3"))
                .log()
                .subscribe();

        System.in.read();
    }

    @Test
    public void testPublishOn() {
        // 当publishOn位于操作之间时，会将上个线程操作的结果发送给Scheduler单独创建的线程中执行。
        Flux.range(1, 100)
                .filter(item -> item % 2 == 0)
                .log()
                .publishOn(Schedulers.newParallel("t1", 4))
                .map(item -> item * 2)
                .log()
                .subscribe();
    }

    @Test
    public void testSubscribeOn() {
        // 当subscribeOn被使用时，那么所有操作都在subscribeOn指定的线程中执行。
        Flux.range(1, 100)
                .subscribeOn(Schedulers.newParallel("t1", 4))
                .filter(item -> item % 2 == 0)
                .log()
                .subscribe();
    }

    @Test
    public void testError() {
        /*
         * 第一步代码：测试异常处理
         *  异常会终止整个流的执行。
         */
        Flux<Integer> flux = Flux.just(1, 3, 4, 5, 6, 0, 7, 8, 9, 10);
        flux.map(i -> 10 / i)
                // 第二步代码：异常之后，返回一个值。
//                .onErrorReturn(-1)\
                // 第三步代码：异常之后，断言式为true，返回一个值。
                .onErrorReturn(err -> {
                    System.out.println("err:" + err);
                    return true;
                }, 100)
                .subscribe(System.out::println,
                        error -> System.out.println("error:" + error));


        /*
         * 第四步代码： 异常之后，执行回调方法。
         * 等效于：
         *      Integer i = null;
                try {
                    i = 10 / 0;
                }catch (Exception e) {
                    i = -1;
                }
         */
//        flux.flatMap(i -> Mono.just(i).map(item -> 10 / i)
//                        .onErrorResume(err -> true, k -> Mono.just(-1)))
//                .subscribe(System.out::println);


        /*
         * 第五步代码： 异常之后，执行回调方法。
         * 等效于：
         *      Integer i = null;
                try {
                    i = 10 / 0;
                }catch (Exception e) {
                    print(错误日志记录)
                    throw e;
                }
         */
        flux.map(i -> 10 / i)
                .doOnError(err -> System.out.println("出错"))
                .subscribe(System.out::println, err -> System.out.println(err));


         /*
         * 第六步代码： 异常之后，总是执行的内容。
         * 等效于：
         *      Integer i = null;
                try {
                    i = 10 / 0;
                }catch (Exception e) {
                    throw e;
                }finally {

                }
         */
        flux.map(i -> 10 / i)
                .doFinally(signalType -> System.out.println("信号。。。。" + signalType))
                .subscribe(System.out::println, err -> System.out.println(err));

    }

    @Test
    public void testSinksOne() {
        Sinks.One<Integer> sinkOne = Sinks.one();

        // 订阅内容
        Mono<Integer> mono = sinkOne.asMono();
        mono.log()
                .subscribe();

        // 发送数据
//        sinkOne.tryEmitValue(10); // 发送一个数据
//        sinkOne.tryEmitValue(20); // 发送一个数据（无用）
        sinkOne.emitValue(10,
                (signalType, emitResult) -> {
                    System.out.println(signalType);
                    System.out.println(emitResult);
                    return false;
                });

        // 发送第二次将触发失败，因为上面已经发送了一个数据
        sinkOne.emitValue(10,
                (signalType, emitResult) -> {
                    System.out.println("发送信号类型：" + signalType);
                    System.out.println("发送结果:" + emitResult);
                    return true; // true时，允许重发，false时，不允许重发
                });
    }

    @Test
    public void testSinksMany() throws IOException, InterruptedException {
        /*
         * 单播：一次只发送一个数据。可以当作是一个Producer
         */
        Sinks.Many<Object> unicastMany = Sinks.many()
                .unicast().onBackpressureBuffer();
        // 订阅
//        unicastMany.asFlux().log()
//                .subscribe(System.out::println);

        // 执行时，会报出UnicastProcessor allows only a single Subscriber的错误
//        new Thread(() -> {
//            unicastMany.asFlux().log()
//                    .subscribe(System.out::println);
//        }).start();

        // 睡眠1秒
//        TimeUnit.SECONDS.sleep(1);

        new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                unicastMany.tryEmitNext("我是 " + i);
            }
        }).start();

        /*
         * 多播：一次发送多个数据
         */
        Sinks.Many<Object> multicastMany = Sinks.many()
                .multicast().onBackpressureBuffer();

        // 订阅者1
        new Thread(() -> {
            multicastMany.asFlux()
                    // 改变下游的执行线程:从main线程 -> t2线程
                    .publishOn(Schedulers.newSingle("t2"))
                    .log()
                    .subscribe(i -> {
                        System.out.println(Thread.currentThread().getName() + "收到数据：" + i);
                    });
        }, "t1").start();
        // 订阅者2
        new Thread(() -> {
            multicastMany.asFlux().log().subscribe();
        }, "t2").start();

        // 睡眠1秒，确保订阅者1和订阅者2已经启动
        TimeUnit.SECONDS.sleep(1);

        // 手动发送数据
        for (int i = 0; i < 5; i++) {
//            multicastMany.tryEmitNext(i);
//            multicastMany.emitNext(i, (signalType, emitResult) -> {
//                System.out.println("接收到的信号类型：" + signalType);
//                System.out.println("接收到的传递的结果：" + emitResult);
//                return false; // 不允许重发
//            });
        }


        /*
         * 重试：发送失败之后进行重试
         */
        Sinks.Many<Object> replay = Sinks.many().replay()
                //.limit(5); 重发N个数据
                //.all(); 重复所有历史数据
                .latest(); //重发最新的数据

        Flux<Object> flux = replay.asFlux();
        // 订阅者1
        flux.log().subscribe();
        new Thread(() -> {
            // 订阅者2隔1s后订阅，是否会收到信息
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // 订阅者2
            flux.log().subscribe();
        }, "t2").start();

        for (int i = 0; i < 10; i++) {
            replay.tryEmitNext(i);
        }

        System.in.read();
    }

}
